package com.cloud.message.rocketmq;

import com.cloud.common.util.JsonUtils;
import com.cloud.message.entity.MessagePending;
import com.cloud.message.support.MessagePusher;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class RocketMQReceiver {

    private static final String TOPIC_MESSAGE_LOWER = "MessageLower";
    private static final String TOPIC_MESSAGE_NORMAL = "MessageNormal";
    private static final String TOPIC_MESSAGE_HIGHER = "MessageHigher";
    private static final String TOPIC_MESSAGE_PENDING = "MessagePending";

    @Autowired
    private MessagePusher messagePusher;

    private void pushMessage(String topic, String message) {
        try {
            log.debug("topic={} message >>> {}", topic, message);
            if (topic.equals(TOPIC_MESSAGE_PENDING)) {
                messagePusher.pushPending(JsonUtils.parseObject(message, MessagePending.class));
            } else {
                messagePusher.pushMessage(JsonUtils.parseObject(message, MQMessage.class));
            }
        } catch (Exception e) {
            log.error("push failure message >>> " + message, e);
        }
    }

    @Component
    @RocketMQMessageListener(topic = TOPIC_MESSAGE_LOWER, consumerGroup = TOPIC_MESSAGE_LOWER)
    class MessageLowerReceiver implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            pushMessage(TOPIC_MESSAGE_LOWER, message);
        }
    }

    @Component
    @RocketMQMessageListener(topic = TOPIC_MESSAGE_NORMAL, consumerGroup = TOPIC_MESSAGE_NORMAL)
    class MessageNormalReceiver implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            pushMessage(TOPIC_MESSAGE_NORMAL, message);
        }
    }

    @Component
    @RocketMQMessageListener(topic = TOPIC_MESSAGE_HIGHER, consumerGroup = TOPIC_MESSAGE_HIGHER)
    class MessageHigherReceiver implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            pushMessage(TOPIC_MESSAGE_HIGHER, message);
        }
    }

    @Component
    @RocketMQMessageListener(topic = TOPIC_MESSAGE_PENDING, consumerGroup = TOPIC_MESSAGE_PENDING)
    class MessagePendingReceiver implements RocketMQListener<String> {
        @Override
        public void onMessage(String message) {
            pushMessage(TOPIC_MESSAGE_PENDING, message);
        }
    }

}


